package com.ultrapower.request;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;

import com.sematext.hbase.wd.AbstractRowKeyDistributor;
import com.ultrapower.util.AgentStatType;
import com.ultrapower.util.AgentStatUtils;
import com.ultrapower.util.BytesUtils;
import com.ultrapower.util.CommonUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.rowkey.RowKeyDistributorFactory;

/**
 * 
 * @author wangy
 *
 */
public class AgentStatTable extends BaseRequest {

	protected static final int AGENT_STAT_VER2_NUM_PARTITIONS = 32;

	private AbstractRowKeyDistributor rowKeyDistributor = RowKeyDistributorFactory.createRowKeyDistributor(0, 33, 64);

	@Override
	void execute() {
		TableName agentStatTableName = this.getTableName(HBaseTables.AGENT_STAT_VER2_STR);
		this.getHbaseTemplate().find(agentStatTableName, this.createScan(),rowKeyDistributor, (ResultScanner scanner) -> {
			scanner.forEach(result -> {
				if (result.isEmpty()) {
		            return;
		        }
		        final byte[] distributedRowKey = result.getRow();
		        System.out.println(getRowKeyStr(distributedRowKey));
		        for (Cell cell : result.rawCells()) {
		        	//TODO
		        	System.out.println("//todo");
		        	System.out.println("\t"+cell);
		        }
			});
			return "";
		});
	}

	@Override
	protected Scan createScan() {
		AgentStatType agentStatType = AgentStatType.CPU_LOAD;
		return createScan(agentStatType, agentId, from, to);
	}

	private Scan createScan(AgentStatType agentStatType, String agentId, long from, long to) {
		long scanRange = to - from;
		long expectedNumRows = ((scanRange - 1) / HBaseTables.AGENT_STAT_TIMESPAN_MS) + 1;
		if (from != AgentStatUtils.getBaseTimestamp(from)) {
			expectedNumRows++;
		}
		if (expectedNumRows > MAX_SCAN_CACHE_SIZE) {
			return this.createScan(agentStatType, agentId, from, to, MAX_SCAN_CACHE_SIZE);
		} else {
			return this.createScan(agentStatType, agentId, from, to, (int) expectedNumRows);
		}
	}

	public Scan createScan(AgentStatType agentStatType, String agentId, long startTimestamp, long endTimestamp,
			int scanCacheSize) {
		byte[] startRowKey = this.encodeRowKey(agentId, agentStatType, AgentStatUtils.getBaseTimestamp(endTimestamp));
		byte[] endRowKey = this.encodeRowKey(agentId, agentStatType,
				AgentStatUtils.getBaseTimestamp(startTimestamp) - HBaseTables.AGENT_STAT_TIMESPAN_MS);
		Scan scan = new Scan(startRowKey, endRowKey);
		scan.setCaching(scanCacheSize);
		scan.setId("AgentStat_" + agentStatType);
		scan.addFamily(HBaseTables.AGENT_STAT_CF_STATISTICS);
		return scan;

	}

	public byte[] encodeRowKey(String agentId, AgentStatType agentStatType, long baseTimestamp) {

		byte[] bAgentId = BytesUtils.toBytes(agentId);
		byte[] bStatType = new byte[] { agentStatType.getRawTypeCode() };
		byte[] rowKey = new byte[HBaseTables.AGENT_NAME_MAX_LEN + bStatType.length + BytesUtils.LONG_BYTE_LENGTH];

		BytesUtils.writeBytes(rowKey, 0, bAgentId);
		BytesUtils.writeBytes(rowKey, HBaseTables.AGENT_NAME_MAX_LEN, bStatType);
		BytesUtils.writeLong(CommonUtils.reverseTimeMillis(baseTimestamp), rowKey,
				HBaseTables.AGENT_NAME_MAX_LEN + bStatType.length);

		return rowKey;
	}

	public String getRowKeyStr(byte[] rowkey) {
		final String agentId = BytesUtils.safeTrim(BytesUtils.toString(rowkey, 0, HBaseTables.AGENT_NAME_MAX_LEN));
		final AgentStatType agentStatType = AgentStatType.fromTypeCode(rowkey[HBaseTables.AGENT_NAME_MAX_LEN]);
		final long reversedBaseTimestamp = BytesUtils.bytesToLong(rowkey,HBaseTables.AGENT_NAME_MAX_LEN + AgentStatType.TYPE_CODE_BYTE_LENGTH);
		final long baseTimestamp = CommonUtils.recoveryTimeMillis(reversedBaseTimestamp);
		return agentId+"\t"+agentStatType.getName()+"\t"+baseTimestamp;
	}

	/**
	 * 
	 * @param args
	 */
	public static void main(String[] args) {
		BaseRequest agentStat=new AgentStatTable();
		agentStat.execute();
	}

}
